package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"time"

	"github.com/google/uuid"

	"watermill-order-demo/internal/db"
	"watermill-order-demo/internal/pubsub"
	"watermill-order-demo/pkg/events"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
)

// Server 订单服务主结构体，包含数据库连接、消息发布器和日志适配器
// db: 订单数据库连接
// publisher: 用于事件发布的 Watermill 消息发布器
// logger: Watermill 日志适配器
type Server struct {
	db        *sql.DB
	publisher message.Publisher // GoChannel publisher the forwarder will use
	logger    watermill.LoggerAdapter
}

// main 订单服务入口，负责初始化数据库、消息通道、启动 HTTP 服务和 Outbox 转发器
func main() {
	// 读取数据库连接和服务监听地址（可通过环境变量覆盖）
	dsn := os.Getenv("DATABASE_URL")
	if dsn == "" {
		dsn = "postgres://demo:demo@localhost:5432/demo?sslmode=disable"
	}
	addr := os.Getenv("ORDER_ADDR")
	if addr == "" {
		addr = ":8080"
	}
	logger := watermill.NewStdLogger(true, true)

	// 初始化数据库连接
	dbConn, err := db.Open(dsn)
	if err != nil {
		log.Fatalf("open db: %v", err)
	}

	// 创建内存消息通道（GoChannel），用于本地演示
	publisher, _ := pubsub.MakeGoChannelPubSub(logger)

	// 构建服务实例
	srv := &Server{
		db:        dbConn,
		publisher: publisher,
		logger:    logger,
	}

	// 启动 Outbox 转发器，定时轮询 outbox 表并发布新消息
	ctx := context.Background()
	go srv.startOutboxForwarder(ctx, 500*time.Millisecond)

	// 注册 HTTP 路由，处理订单创建请求
	http.HandleFunc("/orders", srv.handleCreateOrder)

	fmt.Printf("Order service running at %s\n", addr)
	if err := http.ListenAndServe(addr, nil); err != nil {
		log.Fatalf("listen: %v", err)
	}
}

// handleCreateOrder 处理订单创建请求，写入订单和 outbox 表，并返回订单ID
func (s *Server) handleCreateOrder(w http.ResponseWriter, r *http.Request) {
	// 解析请求体，获取订单信息
	var req struct {
		OrderID string          `json:"order_id"`
		UserID  string          `json:"user_id"`
		Items   []events.Item   `json:"items"`
		Total   float64         `json:"total"`
		Extra   json.RawMessage `json:"extra,omitempty"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "bad request: "+err.Error(), http.StatusBadRequest)
		return
	}
	if req.OrderID == "" {
		req.OrderID = "order-" + uuid.NewString()
	}
	orderEvt := events.NewOrderCreated(req.OrderID, req.UserID, req.Items, req.Total)

	// 开启数据库事务，插入订单和 outbox 消息
	tx, err := s.db.Begin()
	if err != nil {
		http.Error(w, "tx begin: "+err.Error(), http.StatusInternalServerError)
		return
	}
	defer tx.Rollback()

	// 插入订单表
	orderPayload, _ := json.Marshal(orderEvt)
	if _, err := tx.Exec(`INSERT INTO orders (id, user_id, payload, total) VALUES ($1, $2, $3, $4)`,
		orderEvt.OrderID, orderEvt.UserID, orderPayload, orderEvt.Total); err != nil {
		http.Error(w, "insert order: "+err.Error(), http.StatusInternalServerError)
		return
	}

	// 插入 outbox 表（事件消息）
	msg, err := events.EncodeOrderCreated(orderEvt)
	if err != nil {
		http.Error(w, "encode msg: "+err.Error(), http.StatusInternalServerError)
		return
	}
	if _, err := tx.Exec(`INSERT INTO outbox (topic, payload, metadata) VALUES ($1, $2, $3)`,
		"orders.created", msg.Payload, "{}"); err != nil {
		http.Error(w, "insert outbox: "+err.Error(), http.StatusInternalServerError)
		return
	}

	if err := tx.Commit(); err != nil {
		http.Error(w, "tx commit: "+err.Error(), http.StatusInternalServerError)
		return
	}

	// 返回订单ID
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusAccepted)
	_ = json.NewEncoder(w).Encode(map[string]string{"order_id": orderEvt.OrderID})
}

// startOutboxForwarder 定时轮询 outbox 表，发布未处理的消息，并标记为已处理
func (s *Server) startOutboxForwarder(ctx context.Context, interval time.Duration) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.processPendingOutbox(ctx)
		}
	}
}

// processPendingOutbox 处理 outbox 表中的未处理消息，发布事件并更新状态
func (s *Server) processPendingOutbox(ctx context.Context) {
	// 查询未处理的 outbox 消息
	rows, err := s.db.Query(`SELECT id, topic, payload FROM outbox WHERE processed = false ORDER BY created_at LIMIT 50`)
	if err != nil {
		s.logger.Error("query outbox", err, nil)
		return
	}
	defer rows.Close()

	type row struct {
		id      string
		topic   string
		payload []byte
	}
	var batch []row
	for rows.Next() {
		var r row
		if err := rows.Scan(&r.id, &r.topic, &r.payload); err != nil {
			s.logger.Error("scan outbox", err, nil)
			return
		}
		batch = append(batch, r)
	}
	if len(batch) == 0 {
		return
	}

	// 顺序发布每条消息
	for _, r := range batch {
		msg := message.NewMessage(uuid.NewString(), r.payload)
		msg.Metadata.Set("outbox_id", r.id)
		// 发布消息
		if err := s.publisher.Publish(r.topic, msg); err != nil {
			s.logger.Error("publish outbox", err, map[string]interface{}{
				"topic": r.topic,
				"id":    r.id,
			})
			continue
		}
		// 标记为已处理
		if _, err := s.db.Exec(`UPDATE outbox SET processed = true, processed_at = now() WHERE id = $1`, r.id); err != nil {
			s.logger.Error("mark outbox processed", err, map[string]interface{}{"id": r.id})
		}
		s.logger.Info("outbox forwarded", map[string]interface{}{"id": r.id, "topic": r.topic})
	}
}
